//! 文件存储实现
//!
//! 本模块提供了基于文件系统的持久化存储实现。
//! 使用文件系统作为底层存储介质，支持以下特性：
//!
//! - **持久化存储**: 数据以文件形式保存在磁盘上，服务重启后数据不丢失
//! - **键值存储**: 支持任意字符串作为键，二进制数据作为值
//! - **批量操作**: 支持批量读写、删除操作，提高性能
//! - **前缀扫描**: 支持基于键前缀的数据扫描
//! - **原子操作**: 单个文件操作具有原子性，确保数据一致性
//! - **高性能**: 使用异步IO操作，基于Tokio实现高性能文件访问
//!
//! ## 存储结构
//!
//! 数据以以下结构存储在文件系统中：
//! ```
//! <root_path>/
//!   ├── <hex_encoded_key>.dat    # 存储实际数据
//!   └── ...
//! ```
//!
//! ## 键编码策略
//!
//! 为了支持任意字符串作为文件名，使用以下编码策略：
//! - 使用十六进制编码将键转换为安全的文件名
//! - 支持任意Unicode字符作为键名
//! - 避免特殊字符导致的文件系统问题
//!
//! ## 使用示例
//!
//! ```rust
//! use nacos_storage::file::FileStorage;
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
//!     // 创建文件存储实例
//!     let storage = FileStorage::new("/tmp/nacos_data").await?;
//!     
//!     // 存储数据
//!     storage.put("service:user", b"user data").await?;
//!     
//!     // 读取数据
//!     let data = storage.get("service:user").await?;
//!     assert_eq!(data, Some(b"user data".to_vec()));
//!     
//!     Ok(())
//! }
//! ```

use async_trait::async_trait;
use nacos_core::{Storage, Result, NacosError};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use tokio::fs;
use tracing::{debug, trace, warn};

/// 文件存储实现
///
/// `FileStorage` 提供了基于文件系统的键值存储功能。
/// 所有数据以文件形式存储在指定的根目录下，每个键对应一个文件。
#[derive(Debug, Clone)]
pub struct FileStorage {
    /// 存储根目录路径
    /// 
    /// 所有数据文件都存储在这个目录下，结构为：
    /// `<root_path>/<hex_encoded_key>.dat`
    root_path: PathBuf,
}

impl FileStorage {
    /// 创建新的文件存储实例
    ///
    /// # 参数
    /// - `root_path`: 存储根目录路径
    ///
    /// # 返回值
    /// - `Ok(FileStorage)`: 成功创建的文件存储实例
    /// - `Err(NacosError)`: 创建目录失败时的错误
    ///
    /// # 行为
    /// 如果指定的根目录不存在，会自动创建该目录及其所有父目录。
    /// 使用异步IO操作确保高性能。
    pub async fn new<P: AsRef<Path>>(root_path: P) -> Result<Self> {
        let root_path = root_path.as_ref().to_path_buf();
        trace!("Initializing FileStorage at {:?}", root_path);
        if !root_path.exists() {
            debug!("Root directory {:?} does not exist, creating it.", root_path);
            fs::create_dir_all(&root_path).await
                .map_err(|e| NacosError::Storage(format!("Failed to create directory: {}", e)))?;
        }
        Ok(Self { root_path })
    }

    /// 将键转换为安全的文件路径
    ///
    /// # 参数
    /// - `key`: 要转换的键字符串
    ///
    /// # 返回值
    /// - `PathBuf`: 对应的文件路径
    ///
    /// # 编码策略
    /// 使用十六进制编码将任意字符串转换为安全的文件名，
    /// 避免特殊字符导致的文件系统问题。
    ///
    /// # 示例
    /// ```
    /// let storage = FileStorage::new("/tmp/data").await.unwrap();
    /// let path = storage.key_to_path("user:123");
    /// // 结果为: /tmp/data/757365723a313233.dat
    /// ```
    fn key_to_path(&self, key: &str) -> PathBuf {
        let safe_key = hex::encode(key);
        self.root_path.join(format!("{}.dat", safe_key))
    }

    /// 将文件路径转换回原始键
    ///
    /// # 参数
    /// - `path`: 文件路径
    ///
    /// # 返回值
    /// - `Some(String)`: 成功解码的原始键
    /// - `None`: 解码失败（路径格式不正确或编码错误）
    ///
    /// # 解码过程
    /// 1. 提取文件名（去掉.dat后缀）
    /// 2. 将十六进制字符串解码为原始字节
    /// 3. 将字节转换为UTF-8字符串
    fn path_to_key(&self, path: &Path) -> Option<String> {
        path.file_stem()
            .and_then(|s| s.to_str())
            .and_then(|s| hex::decode(s).ok())
            .and_then(|bytes| String::from_utf8(bytes).ok())
    }
}

#[async_trait]
impl Storage for FileStorage {
    /// 根据键获取对应的值
    ///
    /// # 参数
    /// - `key`: 要获取的键字符串
    ///
    /// # 返回值
    /// - `Ok(Some(Vec<u8>))`: 成功获取到数据
    /// - `Ok(None)`: 键不存在
    /// - `Err(NacosError)`: 读取文件失败
    ///
    /// # 行为
    /// 1. 将键转换为对应的文件路径
    /// 2. 异步读取文件内容
    /// 3. 处理各种错误情况
    ///
    /// # 错误处理
    /// - 文件不存在：返回Ok(None)
    /// - 其他IO错误：返回StorageError
    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
        trace!(key, "Getting key from file");
        let path = self.key_to_path(key);
        match fs::read(&path).await {
            Ok(data) => {
                trace!(key, path = ?path, "Successfully read file");
                Ok(Some(data))
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                trace!(key, path = ?path, "File not found");
                Ok(None)
            }
            Err(e) => {
                warn!(key, path = ?path, error = %e, "Failed to read file");
                Err(NacosError::Storage(format!("Failed to read file: {}", e)))
            }
        }
    }

    /// 存储键值对
    ///
    /// # 参数
    /// - `key`: 键字符串
    /// - `value`: 要存储的二进制数据
    ///
    /// # 返回值
    /// - `Ok(())`: 存储成功
    /// - `Err(NacosError)`: 存储失败
    ///
    /// # 行为
    /// 1. 将键转换为对应的文件路径
    /// 2. 确保父目录存在（自动创建）
    /// 3. 异步写入文件内容
    /// 4. 处理各种错误情况
    ///
    /// # 原子性
    /// 单个文件的写入操作是原子的，要么完全成功，要么完全失败。
    /// 如果写入过程中发生错误，文件内容不会被部分修改。
    async fn put(&self, key: &str, value: &[u8]) -> Result<()> {
        trace!(key, value_len = value.len(), "Putting key to file");
        let path = self.key_to_path(key);
        if let Some(parent) = path.parent() {
            if !parent.exists() {
                fs::create_dir_all(parent).await.map_err(|e| {
                    warn!(key, path = ?path, error = %e, "Failed to create parent directory");
                    NacosError::Storage(format!("Failed to create parent directory: {}", e))
                })?;
            }
        }
        fs::write(&path, value).await
            .map_err(|e| {
                warn!(key, path = ?path, error = %e, "Failed to write file");
                NacosError::Storage(format!("Failed to write file: {}", e))
            })?;
        trace!(key, path = ?path, "Successfully wrote file");
        Ok(())
    }

    /// 删除指定键对应的数据
    ///
    /// # 参数
    /// - `key`: 要删除的键字符串
    ///
    /// # 返回值
    /// - `Ok(())`: 删除成功（包括键不存在的情况）
    /// - `Err(NacosError)`: 删除失败
    ///
    /// # 行为
    /// 1. 将键转换为对应的文件路径
    /// 2. 异步删除文件
    /// 3. 处理各种错误情况
    ///
    /// # 幂等性
    /// 删除不存在的键会返回Ok(())，不会报错。
    async fn delete(&self, key: &str) -> Result<()> {
        trace!(key, "Deleting key from file");
        let path = self.key_to_path(key);
        match fs::remove_file(&path).await {
            Ok(_) => {
                trace!(key, path = ?path, "Successfully deleted file");
                Ok(())
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                trace!(key, path = ?path, "File not found for deletion, considering it a success");
                Ok(())
            }
            Err(e) => {
                warn!(key, path = ?path, error = %e, "Failed to delete file");
                Err(NacosError::Storage(format!("Failed to delete file: {}", e)))
            }
        }
    }

    /// 扫描指定前缀的所有键值对
    ///
    /// # 参数
    /// - `prefix`: 键前缀字符串
    ///
    /// # 返回值
    /// - `Ok(Vec<(String, Vec<u8>)>>`: 匹配前缀的所有键值对
    /// - `Err(NacosError)`: 扫描失败
    ///
    /// # 行为
    /// 1. 遍历存储目录下的所有文件
    /// 2. 将文件名解码为原始键
    /// 3. 检查键是否以指定前缀开头
    /// 4. 读取匹配文件的内容
    /// 5. 返回所有匹配的键值对
    ///
    /// # 性能
    /// 这是一个线性扫描操作，时间复杂度为O(n)，其中n是存储中的总键数。
    /// 对于大量数据，建议使用更高效的索引机制。
    async fn scan(&self, prefix: &str) -> Result<Vec<(String, Vec<u8>)>> {
        trace!(prefix, "Scanning files with prefix");
        let mut result = Vec::new();
        let mut entries = fs::read_dir(&self.root_path).await
            .map_err(|e| NacosError::Storage(format!("Failed to read directory: {}", e)))?;

        while let Some(entry) = entries.next_entry().await.ok().flatten() {
            let path = entry.path();
            if path.is_file() {
                if let Some(key) = self.path_to_key(&path) {
                    if key.starts_with(prefix) {
                        match fs::read(&path).await {
                            Ok(data) => result.push((key, data)),
                            Err(e) => warn!(path = ?path, error = %e, "Failed to read file during scan"),
                        }
                    }
                }
            }
        }
        debug!(prefix, count = result.len(), "Scan completed");
        Ok(result)
    }

    /// 批量获取多个键对应的值
    ///
    /// # 参数
    /// - `keys`: 要获取的键列表
    ///
    /// # 返回值
    /// - `Ok(HashMap<String, Vec<u8>>)`: 成功获取的键值对映射
    ///   - 只包含实际存在的键值对
    ///   - 不存在的键会被忽略
    /// - `Err(NacosError)`: 批量获取失败
    ///
    /// # 行为
    /// 1. 对输入的每个键调用get方法
    /// 2. 收集所有存在的键值对
    /// 3. 忽略不存在的键
    ///
    /// # 性能
    /// 这是串行操作，每个键的获取操作按顺序执行。
    /// 对于大量键，可以考虑并行获取以提高性能。
    async fn batch_get(&self, keys: &[String]) -> Result<HashMap<String, Vec<u8>>> {
        trace!(num_keys = keys.len(), "Batch getting keys from files");
        let mut result = HashMap::new();
        for key in keys {
            if let Some(data) = self.get(key).await? {
                result.insert(key.clone(), data);
            }
        }
        debug!(found = result.len(), total = keys.len(), "Batch get completed");
        Ok(result)
    }

    /// 批量存储多个键值对
    ///
    /// # 参数
    /// - `data`: 要存储的键值对映射
    ///
    /// # 返回值
    /// - `Ok(())`: 批量存储成功
    /// - `Err(NacosError)`: 存储过程中遇到错误
    ///
    /// # 行为
    /// 1. 对输入的每个键值对调用put方法
    /// 2. 按顺序存储每个键值对
    /// 3. 如果中途遇到错误，立即停止并返回错误
    ///
    /// # 原子性
    /// 这不是原子操作。如果中途失败，部分数据可能已成功存储。
    /// 调用者需要处理部分成功的情况。
    async fn batch_put(&self, data: &HashMap<String, Vec<u8>>) -> Result<()> {
        trace!(num_items = data.len(), "Batch putting items to files");
        for (key, value) in data {
            self.put(key, value).await?;
        }
        debug!(num_items = data.len(), "Batch put completed");
        Ok(())
    }

    /// 批量删除多个键对应的数据
    ///
    /// # 参数
    /// - `keys`: 要删除的键列表
    ///
    /// # 返回值
    /// - `Ok(())`: 批量删除成功
    /// - `Err(NacosError)`: 删除过程中遇到错误
    ///
    /// # 行为
    /// 1. 对输入的每个键调用delete方法
    /// 2. 按顺序删除每个键对应的数据
    /// 3. 如果中途遇到错误，立即停止并返回错误
    /// 4. 不存在的键会被忽略
    ///
    /// # 幂等性
    /// 删除不存在的键不会报错，会被忽略。
    /// 这不是原子操作，部分删除可能已成功执行。
    async fn batch_delete(&self, keys: &[String]) -> Result<()> {
        trace!(num_keys = keys.len(), "Batch deleting keys from files");
        for key in keys {
            self.delete(key).await?;
        }
        debug!(num_keys = keys.len(), "Batch delete completed");
        Ok(())
    }

    /// 检查指定键是否存在
    ///
    /// # 参数
    /// - `key`: 要检查的键字符串
    ///
    /// # 返回值
    /// - `Ok(true)`: 键存在
    /// - `Ok(false)`: 键不存在
    /// - `Err(NacosError)`: 检查失败
    ///
    /// # 行为
    /// 直接检查对应的文件是否存在。
    /// 这是一个轻量级操作，不会读取文件内容。
    async fn exists(&self, key: &str) -> Result<bool> {
        trace!(key, "Checking file existence");
        Ok(self.key_to_path(key).exists())
    }

    /// 获取匹配指定模式的键列表
    ///
    /// # 参数
    /// - `pattern`: 模式字符串，支持通配符
    ///   - 以`*`结尾：匹配前缀（如"user:*"匹配所有以"user:"开头的键）
    ///   - 精确匹配：匹配完全相等的键
    ///
    /// # 返回值
    /// - `Ok(Vec<String>)`: 匹配的键列表
    /// - `Err(NacosError)`: 扫描失败
    ///
    /// # 行为
    /// 1. 遍历存储目录下的所有文件
    /// 2. 将文件名解码为原始键
    /// 3. 根据模式匹配键
    /// 4. 返回所有匹配的键
    ///
    /// # 性能
    /// 这是一个线性扫描操作，时间复杂度为O(n)。
    /// 对于大量数据，建议使用专门的索引机制。
    async fn keys(&self, pattern: &str) -> Result<Vec<String>> {
        trace!(pattern, "Getting keys from files with pattern");
        let mut result = Vec::new();
        let mut entries = fs::read_dir(&self.root_path).await
            .map_err(|e| NacosError::Storage(format!("Failed to read directory: {}", e)))?;
        while let Some(entry) = entries.next_entry().await.ok().flatten() {
            let path = entry.path();
            if path.is_file() {
                if let Some(key) = self.path_to_key(&path) {
                    let mut pattern_prefix = pattern.to_string();
                    let is_wildcard = if pattern.ends_with('*') {
                        pattern_prefix.pop();
                        true
                    } else {
                        false
                    };

                    if is_wildcard {
                        if key.starts_with(&pattern_prefix) {
                            result.push(key);
                        }
                    } else if key == pattern {
                        result.push(key);
                    }
                }
            }
        }
        debug!(pattern, count = result.len(), "Keys scan completed");
        Ok(result)
    }

    /// 清空所有存储的数据
    ///
    /// # 返回值
    /// - `Ok(())`: 清空成功
    /// - `Err(NacosError)`: 清空失败
    ///
    /// # 行为
    /// 1. 遍历存储目录下的所有文件
    /// 2. 删除所有数据文件
    /// 3. 保留目录结构（只删除文件，不删除目录）
    ///
    /// # 警告
    /// 这是一个破坏性操作，会永久删除所有存储的数据。
    /// 调用前请确保已备份重要数据。
    ///
    /// # 原子性
    /// 这不是原子操作。如果中途失败，部分文件可能已被删除。
    /// 错误处理需要考虑部分删除的情况。
    async fn clear(&self) -> Result<()> {
        trace!("Clearing all files from storage");
        let mut count = 0;
        let mut entries = fs::read_dir(&self.root_path).await
            .map_err(|e| NacosError::Storage(format!("Failed to read directory: {}", e)))?;
        while let Some(entry) = entries.next_entry().await.ok().flatten() {
            if entry.path().is_file() {
                if fs::remove_file(entry.path()).await.is_ok() {
                    count += 1;
                }
            }
        }
        debug!(count, "Clear completed");
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    #[tokio::test]
    async fn test_file_storage_put_get() {
        let dir = tempdir().unwrap();
        let storage = FileStorage::new(dir.path()).await.unwrap();
        let key = "test:key/with/slashes";
        let value = b"hello world".to_vec();

        storage.put(key, &value).await.unwrap();
        let retrieved = storage.get(key).await.unwrap().unwrap();
        assert_eq!(value, retrieved);
    }

    #[tokio::test]
    async fn test_file_storage_delete() {
        let dir = tempdir().unwrap();
        let storage = FileStorage::new(dir.path()).await.unwrap();
        let key = "test:key";
        storage.put(key, b"value").await.unwrap();

        storage.delete(key).await.unwrap();
        let retrieved = storage.get(key).await.unwrap();
        assert!(retrieved.is_none());
    }
    
    #[tokio::test]
    async fn test_file_storage_scan() {
        let dir = tempdir().unwrap();
        let storage = FileStorage::new(dir.path()).await.unwrap();
        storage.put("prefix:a", b"1").await.unwrap();
        storage.put("prefix:b", b"2").await.unwrap();
        storage.put("other:c", b"3").await.unwrap();

        let results = storage.scan("prefix:").await.unwrap();
        assert_eq!(results.len(), 2);
        // Make sure we got the right ones
        let keys: Vec<String> = results.into_iter().map(|(k, _)| k).collect();
        assert!(keys.contains(&"prefix:a".to_string()));
        assert!(keys.contains(&"prefix:b".to_string()));
    }

    #[tokio::test]
    async fn test_file_storage_batch_operations() {
        let dir = tempdir().unwrap();
        let storage = FileStorage::new(dir.path()).await.unwrap();
        let mut data = HashMap::new();
        data.insert("key1".to_string(), b"val1".to_vec());
        data.insert("key2".to_string(), b"val2".to_vec());

        storage.batch_put(&data).await.unwrap();
        assert!(storage.exists("key1").await.unwrap());
        assert!(storage.exists("key2").await.unwrap());

        storage.batch_delete(&vec!["key1".to_string()]).await.unwrap();
        assert!(!storage.exists("key1").await.unwrap());
        assert!(storage.exists("key2").await.unwrap());
    }

    #[tokio::test]
    async fn test_file_storage_keys() {
        let dir = tempdir().unwrap();
        let storage = FileStorage::new(dir.path()).await.unwrap();
        storage.put("user:1", b"").await.unwrap();
        storage.put("user:2", b"").await.unwrap();
        storage.put("org:1", b"").await.unwrap();

        let all_keys = storage.keys("*").await.unwrap();
        assert_eq!(all_keys.len(), 3);

        let user_keys = storage.keys("user:*").await.unwrap();
        assert_eq!(user_keys.len(), 2);
    }

    #[tokio::test]
    async fn test_file_storage_clear() {
        let dir = tempdir().unwrap();
        let storage = FileStorage::new(dir.path()).await.unwrap();
        storage.put("a", b"1").await.unwrap();
        storage.put("b", b"2").await.unwrap();

        storage.clear().await.unwrap();
        let all_keys = storage.keys("*").await.unwrap();
        assert!(all_keys.is_empty());
    }
}
